package com.deep.test

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel
import org.apache.spark.mllib.recommendation.Rating

/**
 * @author sw
 * @create 2023-05-29 15:57
 */
object Test14 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("ALS")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val data = sc.textFile("data/als/test.data")
    val ratings = data.map(_.split(',') match { case Array(user, item, rate)
    => Rating(user.toInt, item.toInt, rate.toDouble)
    })
    ratings.foreach { x => println(x) }
    val splits = ratings.randomSplit(Array(0.8, 0.2))
    val training = splits(0)

    val test = splits(1)
    val rank = 10
    val numIterations = 10
    val model = ALS.train(training, rank, numIterations, 0.01)


    //    val model = new ALS()
    //      .setRank(params.rank)
    //      .setIterations(params.numIterations)
    //      .setLambda(params.lambda)
    //      .setImplicitPrefs(params.implicitPrefs)
    //      .setUserBlocks(params.numUserBlocks)
    //      .setProductBlocks(params.numProductBlocks)
    //      .run(training)


    val testUsersProducts = test.map { case Rating(user, product, rate) =>
      (user, product)
    }
    val predictions =
      model.predict(testUsersProducts).map { case Rating(user, product, rate)
      => ((user, product), rate)
      }
    val ratesAndPreds = test.map { case Rating(user, product, rate) =>
      ((user, product), rate)

    }.join(predictions)

    ratesAndPreds.foreach(println)

    val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>

      val err = (r1 - r2)
      err * err

    }.mean()
    println("Mean Squared Error = " + MSE)

  }
}
